package com.zhang.flink.example;

import com.zhang.flink.bean.ClickEvent;
import com.zhang.flink.example.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;

/**
 * @title: 每隔5s统计过去10s访问url  Top3
 * @author: zhang
 * @date: 2022/3/17 08:33
 */
public class UrlCountTopN {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //TODO 2.读取数据，分配水位线和提取事件事件字段
        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<ClickEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                                    @Override
                                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .map(r -> r.url)
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .aggregate(new UrlCountAgg(), new AllWindowFunc())
                .print();

        //TODO 1.执行任务
        env.execute();
    }

    public static class AllWindowFunc extends ProcessAllWindowFunction<ArrayList<Tuple2<String, Long>>, String, TimeWindow> {
        @Override
        public void process(ProcessAllWindowFunction<ArrayList<Tuple2<String, Long>>, String, TimeWindow>.Context context, Iterable<ArrayList<Tuple2<String, Long>>> elements, Collector<String> out) throws Exception {
            ArrayList<Tuple2<String, Long>> urlList = elements.iterator().next();
            StringBuilder result = new StringBuilder("----------------" + new Timestamp(context.window().getStart()) + "----------------\n");
            for (int i = 0; i < 3; i++) {
                Tuple2<String, Long> tuple2 = urlList.get(i);
                result.append("NO." + (i + 1) + " url:" + tuple2.f0 + ",总浏览次数：" + tuple2.f1 + "\n");
            }
            result.append("----------------" + new Timestamp(context.window().getEnd()) + "----------------\n");

            out.collect(result.toString());
        }
    }


    public static class UrlCountAgg implements AggregateFunction<String, HashMap<String, Long>, ArrayList<Tuple2<String, Long>>> {
        @Override
        public HashMap<String, Long> createAccumulator() {
            return new HashMap<>();
        }

        @Override
        public HashMap<String, Long> add(String value, HashMap<String, Long> accumulator) {
            if (accumulator.containsKey(value)) {
                Long count = accumulator.get(value);
                accumulator.put(value, count + 1L);
            } else {
                accumulator.put(value, 1L);
            }
            return accumulator;
        }

        @Override
        public ArrayList<Tuple2<String, Long>> getResult(HashMap<String, Long> accumulator) {
            ArrayList<Tuple2<String, Long>> list = new ArrayList<>();
            for (String key : accumulator.keySet()) {
                list.add(Tuple2.of(key, accumulator.get(key)));
            }

            list.sort(new Comparator<Tuple2<String, Long>>() {
                @Override
                public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                    return o2.f1.intValue() - o1.f1.intValue();
                }
            });
            return list;
        }

        @Override
        public HashMap<String, Long> merge(HashMap<String, Long> a, HashMap<String, Long> b) {
            for (String key : a.keySet()) {
                if (b.containsKey(key)) {
                    Long bCnt = b.get(key);
                    Long aCnt = a.get(key);
                    b.put(key, bCnt + aCnt);
                } else {
                    b.put(key, a.get(key));
                }
            }
            return b;
        }
    }
}
